package com.cn.common;

import com.cn.common.topic.TopicData;
import com.cn.util.RemotingSerializable;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 处理topic-queue路由
 */
public class TopicRouteManager {

    private final ConcurrentMap<String /*topic*/ , List<QueueData>> topicRouteTables = new ConcurrentHashMap<>();

    //集群
    private final Set<String> cluster = new HashSet<>(16);

    //集群下节点
    private final Map<String /*cluster*/, Set<String> /*brokerName*/> brokerName = new HashMap<>(64);

    //节点集(存放所有broker的详细信息)
    private final Map<String /*brokerName*/, BrokerData> nodes = new HashMap<>(128);

    //当前活跃节点
    private final Map<Channel /*addr*/, BrokerEntity> node_tables = new HashMap<>(128);

    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    /**
     * broker节点注册
     *
     * @param brokerEntity
     * @param ctx
     */
    public void registerBroker(BrokerEntity brokerEntity, ChannelHandlerContext ctx) {
        //集群注册
        this.cluster.add(brokerEntity.getCluster_name());
        try {
            if (this.lock.writeLock().tryLock(3000, TimeUnit.MILLISECONDS)) {
                //节点注册
                Set<String> brokerNames = this.brokerName.get(brokerEntity.getCluster_name());
                if (brokerNames == null) {
                    brokerNames = new HashSet<>();
                    this.brokerName.put(brokerEntity.getCluster_name(), brokerNames);
                }
                brokerNames.add(brokerEntity.getName());

                //节点集注册
                try {
                    BrokerData brokerData = this.nodes.get(brokerEntity.getName());
                    if (null == brokerData) {
                        brokerData = new BrokerData();
                        this.nodes.put(brokerEntity.getName(), brokerData);
                    }
                    if (brokerData != null) {
                        brokerData.setCluster(brokerEntity.getCluster_name());
                        brokerData.setBrokerName(brokerEntity.getName());
                        brokerData.getAddr().put(brokerEntity.getBroker_id(), brokerEntity.getAddress());
                    }

                    node_tables.put(ctx.channel(), brokerEntity);

                } finally {
                    this.lock.writeLock().unlock();
                }
            }
//            System.out.println("cluster:" + this.cluster);
//            System.out.println("node_tables:" + this.node_tables);
//            System.out.println("nodes：" + this.nodes);
//            System.out.println("brokerNames:" + this.brokerName);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 取消注册
     *
     * @param ctx
     */
    public void unregister(ChannelHandlerContext ctx) {

        BrokerEntity brokerEntity = node_tables.remove(ctx.channel());
        if (null == brokerEntity) {
            return;
        }
        try {
            //节点集注册
            if (this.lock.writeLock().tryLock(3000, TimeUnit.MILLISECONDS)) {
                try {
                    BrokerData brokerData = this.nodes.get(brokerEntity.getName());
                    if (brokerData != null) {
                        Map<Integer, String> address = brokerData.getAddr();
                        address.remove(brokerEntity.getBroker_id());
                        if (address == null || address.size() == 0) {
                            this.nodes.remove(brokerEntity.getName());
                            Set<String> brokerNames = this.brokerName.get(brokerEntity.getCluster_name());
                            if (brokerNames != null) {

                            }
                            brokerNames.remove(brokerEntity.getName());
                            if (brokerNames == null || brokerNames.size() == 0) {
                                //集群下的节点已全部失效
                                this.brokerName.remove(brokerEntity.getCluster_name());
                            }
                        }
                    }

                } finally {
                    this.lock.writeLock().unlock();
                }
            }

//            System.out.println("cluster:" + this.cluster);
//            System.out.println("node_tables:" + this.node_tables);
//            System.out.println("nodes：" + this.nodes);
//            System.out.println("brokerNames:" + this.brokerName);

        } catch (InterruptedException e) {

        }
    }


    /**
     * 注册topic
     *
     * @param tables
     * @param ct
     */
    public void registerTopic(ConcurrentMap<String, TopicData> tables, ChannelHandlerContext ct) {

        BrokerEntity brokerEntity = this.node_tables.get(ct.channel());
        if (brokerEntity == null) {
            System.out.println("==》");
            return;
        }
        for (String key : tables.keySet()) {
            addQueueData(brokerEntity, tables.get(key));
        }
    }

    public void addQueueData(BrokerEntity brokerEntity, TopicData topicData) {
        List<QueueData> list = topicRouteTables.get(topicData.getTopic());
        QueueData queueData = new QueueData();
        queueData.setTopic(topicData.getTopic());//topic
        queueData.setReadQueue(topicData.getReadNum());//读队列数
        queueData.setWriteQueue(topicData.getWriteNum());//写队列数
        queueData.setBrokerName(brokerEntity.getName());//所在节点

        if (list == null) {
            list = new ArrayList<>();
            list.add(queueData);

            topicRouteTables.put(topicData.getTopic(), list);
        } else {

            Iterator<QueueData> iterator = list.iterator();
            boolean createQueue = true;
            while (iterator.hasNext()) {
                QueueData old = iterator.next();
                //保证是同一个节点
                if (old.getBrokerName().equals(brokerEntity.getName())) {
                    if (old.equals(queueData)) {
                        createQueue = false;
                    } else {
                        iterator.remove();
                    }
                }
            }

            if (createQueue) {
                list.add(queueData);
            }
        }
    }

    /**
     * 拉取topic
     *
     * @param topic
     * @return
     */
    public byte[] filterTopicRouteData(String topic) {
        TopicRouteData topicRouteData = new TopicRouteData();
        List<QueueData> queueDataList = this.topicRouteTables.get(topic);
        if (queueDataList != null) {
            topicRouteData.setQueueData(queueDataList);
            List<BrokerData> brokerDataList = new ArrayList<>();
            for (QueueData queueData : queueDataList) {
                BrokerData brokerData = this.nodes.get(queueData.getBrokerName());
                if (brokerData != null) {
                    brokerDataList.add(brokerData);
                }
            }
            topicRouteData.setBrokerData(brokerDataList);
        } else {
            topicRouteData.setQueueData(new ArrayList<>());
            topicRouteData.setBrokerData(new ArrayList<>());
        }
        return RemotingSerializable.encode(topicRouteData);
    }

    /**
     * 拉取所有的topic
     */
    public byte[] pullAllTopic() {
        TopicRouteData topicRouteData = new TopicRouteData();
        List<QueueData> queueDataList = new ArrayList<>();
        List<BrokerData> brokerDataList = new ArrayList<>();
        for (String topic : this.topicRouteTables.keySet()) {
            List<QueueData> queueData = this.topicRouteTables.get(topic);
            if (queueData != null) {
                queueDataList.addAll(queueData);
            }
            for (QueueData queue : queueData) {
                BrokerData brokerData = this.nodes.get(queue.getBrokerName());
                if (brokerData != null) {
                    brokerDataList.add(brokerData);
                }
            }
        }
        topicRouteData.setBrokerData(brokerDataList);
        topicRouteData.setQueueData(queueDataList);
        return RemotingSerializable.encode(topicRouteData);
    }

}
